package com.kancy.streamplus.channel;

import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.StringUtils;

import java.time.Duration;

/**
 * <p>
 * RetryPollingMessageExceptionHandler
 * <p>
 *
 * @author: kancy
 * @date: 2020/3/23 10:43
 **/
public class RetryPollingMessageExceptionHandler implements PollingMessageExceptionHandler {
    private final PollingChannel pollingChannel;

    public RetryPollingMessageExceptionHandler(PollingChannel pollingChannel) {
        this.pollingChannel = pollingChannel;
    }

    /**
     * 处理异常
     *
     * @param e
     * @param message
     */
    @Override
    public void onException(Message<?> message, Exception e) {
        // 重发消息
        MessageHeaders headers = message.getHeaders();
        Integer maxRetryTimes = headers.get(PollingMessage.MAX_RETRY_TIMES, Integer.class);
        int retryCount = headers.get(PollingMessage.RETRY_COUNT, Integer.class);

        // 最大重试次数限制
        if (retryCount < maxRetryTimes){
            MessageBuilder<?> messageBuilder = MessageBuilder.withPayload(message.getPayload())
                    .copyHeaders(headers)
                    .setHeader(PollingMessage.RETRY_COUNT, ++retryCount)
                    .setHeader(PollingMessage.DELAY, getRetryFrequency(headers))
                    .setHeader(PollingMessage.LAST_QUERY_TIMESTAMP, System.currentTimeMillis() + 500);
            pollingChannel.output().send(messageBuilder.build());
        }
    }

    /**
     * 获取重试频率
     * @param headers
     * @return
     */
    private long getRetryFrequency(MessageHeaders headers) {
        String frequency = headers.get(PollingMessage.FREQUENCY, String.class);
        int retryCount = headers.get(PollingMessage.RETRY_COUNT, Integer.class);
        if (StringUtils.isEmpty(frequency)){
            return 0;
        }
        String[] frequencyArray = frequency.split(",");
        if (retryCount < frequencyArray.length){
            return Duration.parse(frequencyArray[retryCount]).toMillis();
        }else {
            return Duration.parse(frequencyArray[frequencyArray.length - 1]).toMillis();
        }
    }
}
